package com.bjsdzk.collector.one2one;

import com.alibaba.fastjson.JSON;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.OnExceptionContext;
import com.aliyun.openservices.ons.api.SendCallback;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.bean.ProducerBean;
import com.aliyun.openservices.ons.api.exception.ONSClientException;
import com.bjsdzk.collector.config.MqConfig;
import com.bjsdzk.collector.constant.Constant;
import com.bjsdzk.collector.entity.agreement.CallBackCmd;
import com.bjsdzk.collector.entity.agreement.SendCmd;
import com.bjsdzk.collector.entity.model.ExceColleContr;
import com.bjsdzk.collector.entity.model.ParamsSetting;
import com.bjsdzk.collector.serviceImp.DictionariesService;
import com.bjsdzk.collector.serviceImp.ExceColleContrService;
import com.bjsdzk.collector.serviceImp.ParamsSettingService;
import com.bjsdzk.collector.utils.PublicUtils;
import com.bjsdzk.collector.utils.StrUtils;
import com.bjsdzk.collector.utils.TimeUtil;
import com.bjsdzk.collector.utils.Utils;
import com.google.common.primitives.Bytes;
import com.google.gson.Gson;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.AsyncResult;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.datagram.DatagramSocket;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetServerOptions;
import io.vertx.core.net.NetSocket;
import io.vertx.core.shareddata.Counter;
import io.vertx.core.shareddata.SharedData;
import io.vertx.ext.mongo.BulkOperation;
import io.vertx.ext.mongo.MongoClient;
import io.vertx.ext.mongo.UpdateOptions;
import io.vertx.redis.RedisClient;
import lombok.Data;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.lang3.StringUtils;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.env.Environment;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

public class BasetVerticle extends AbstractVerticle {

    private static Logger logger = LoggerFactory.getLogger(BasetVerticle.class);

    protected static int BINARY_1=0x1;
    protected static int BINARY_2=0x1<<1;
    protected static int BINARY_3=0x1<<2;
    protected static int BINARY_4=0x1<<3;
    protected static int BINARY_5=0x1<<4;
    protected static int BINARY_6=0x1<<5;
    protected static int BINARY_7=0x1<<6;
    protected static int BINARY_8=0x1<<7;
    protected static int BINARY_9=0x1<<8;
    protected static int BINARY_10=0x1<<9;
    protected static int BINARY_11=0x1<<10;
    protected static int BINARY_12=0x1<<11;
    protected static int BINARY_13=0x1<<12;
    protected static int BINARY_14=0x1<<13;
    protected static int BINARY_15=0x1<<14;
    protected static int BINARY_16=0x1<<15;
    protected static int BINARY_17=0x1<<16;
    protected static int BINARY_18=0x1<<17;
    protected static int BINARY_19=0x1<<18;
    protected static int BINARY_20=0x1<<19;
    protected static int BINARY_21=0x1<<20;
    protected static int BINARY_22=0x1<<21;
    protected static int BINARY_23=0x1<<22;
    protected static int BINARY_24=0x1<<23;
    protected static int BINARY_25=0x1<<24;
    protected static int BINARY_26=0x1<<25;
    protected static int BINARY_27=0x1<<26;
    protected static int BINARY_28=0x1<<27;
    protected static int BINARY_29=0x1<<28;
    protected static int BINARY_30=0x1<<29;
    protected static int BINARY_31=0x1<<30;
    protected static int BINARY_32=0x1<<31;

    protected Integer port;
    protected String dtuType;
    protected String heartPackage;
    protected String companyId;
    protected String hardwareType;
    //    protected Map<String,String> deviceTypeList;
    protected EventBus eventBus;
    protected SharedData sharedData;

    protected List<String> deviceList;

    protected volatile boolean lineStatus = false;
    protected AtomicInteger conectCount = new AtomicInteger(0);

    protected Map<String, Integer> lineStatusMap = new ConcurrentHashMap<>(4096);
//    protected Map<String, Long> devicesLastDataTime = new ConcurrentHashMap<>(4096);
//    protected Map<String, DeviceInfoUdp> remoteInfoMap =new ConcurrentHashMap();
    protected Map<String, DeviceInfoUdp> devicesInfo = new ConcurrentHashMap<>(4096);

    protected Map<String, List<Long>> timerMap = new HashMap<>();
    protected Map<String, List<Long>> timerOther = new HashMap<>();




    protected List<Long> timerMapUdp= new ArrayList<>();
    protected List<Long> timerOtherUdp = new ArrayList<>();

    protected RedisClient redis;
    protected MongoClient mongoClient;
    protected ExceColleContrService exceColleContrService;
    protected ParamsSettingService paramsSettingService;
    protected DictionariesService dictionariesService;
    protected ProducerBean producer;
    protected MqConfig mqConfig;
    protected Environment env;

    private AtomicInteger count=new AtomicInteger(0);

    private Map<String,Integer> cmdCallBackFailCount=new ConcurrentHashMap<>();
//    protected Map<Integer, ParamsSetting> paramsSettingAll;
//    protected Map<Integer, ExceColleContr> exceColleall;


//    private AtomicInteger saveintc=new AtomicInteger(0);

    public BasetVerticle() {
    }

    protected void serverListen(NetServer server, RedisClient redis, String key) {
        server.listen(port, netServerAsyncResult -> {
            if (netServerAsyncResult.succeeded()) {
                redis.hset(key, "deploymentID", deploymentID(), result -> {
                    if (result.failed()) {
                        logger.error("deployment fail ==> port:" + port + " ==> " + result.cause().getMessage());
                    }
                });

            } else {
                logger.error("start error port:{},errorMessage:{}",port , netServerAsyncResult.cause().getMessage());
                vertx.undeploy(deploymentID());
                eventBus.send(Constant.MOD_START_FAIL, config());
            }

        });
    }


    protected NetServer getNetServer() {
        NetServerOptions options = new NetServerOptions()/* .setReceiveBufferSize(8192) */.setTcpNoDelay(true)
                .setSoLinger(100).setPort(port).setTcpKeepAlive(true);
        return vertx.createNetServer(options);
    }

    protected void getInit() {
        eventBus = vertx.eventBus();
        sharedData = vertx.sharedData();
        port = Integer.parseInt(config().getString("port"));
        dtuType = config().getString("dtuType");
        companyId = config().getString("companyId");
        heartPackage = config().getString("heartPackage");
        hardwareType = config().getString("hardwareType");

    }

    protected void defaultStart(NetSocket netSocket, Map<String, Long> keepLiveMap, PriorityBlockingQueue<SendCmd> sendQueue, ConcurrentLinkedQueue<CallBackCmd> returnQueue
    ) {
        lineStatus = true;
        String handlerID = netSocket.writeHandlerID();
        timerMap.put(handlerID, new ArrayList<>());
        timerOther.put(handlerID, new ArrayList<>());
        keepLiveMap.put("alive", System.currentTimeMillis());

        calcCount(sharedData, true);
        upLine(keepLiveMap, handlerID);

        checkLine(30000, 180000, keepLiveMap, netSocket, timerOther.get(handlerID));
        startConsumer(netSocket, timerOther.get(handlerID), sendQueue, returnQueue, lineStatus);
        closeHandler(netSocket, timerMap, timerOther, sharedData);
    }

    protected void checkLine(long time, long timeout, Map<String, Long> keepLiveMap, NetSocket netSocket, List<Long> timerOther) {
        long overtimeId = vertx.setPeriodic(time, t -> {
            if (keepLiveMap.get("alive") != null
                    && (System.currentTimeMillis() - keepLiveMap.get("alive") > timeout) && !netSocket.writeQueueFull()) {
                timeOutClose(netSocket, keepLiveMap);
            }
        });
        timerOther.add(overtimeId);
    }

    public JsonObject returnResult(Object returnData) {
        return new JsonObject().put("companyId", companyId).put("port", port).put("dtuType", dtuType).put("hardwareType", hardwareType).put("createTime", TimeUtil.getCurrentDatetime()).put("data", returnData);
    }

    public JsonObject returnResult(String deviceId, Object returnData) {
        return returnResult(companyId,deviceId,returnData);
    }

    public JsonObject returnResult(String companyId,String deviceId, Object returnData) {
        return returnResult(companyId,deviceId,returnData,TimeUtil.getCurrentDatetime());
    }

    public JsonObject returnResult(String companyId,String deviceId, Object returnData,String dataTime) {
        if (devicesInfo.get(deviceId)!=null){
            return new JsonObject().put("companyId", companyId).put("source",devicesInfo.get(deviceId).getSource()).put("port", port).put("dtuType", dtuType).put("hardwareType", hardwareType).put("deviceId", deviceId).put("createTime", dataTime).put("data", returnData);
        }else {
            return new JsonObject().put("companyId", companyId).put("port", port).put("dtuType", dtuType).put("hardwareType", hardwareType).put("deviceId", deviceId).put("createTime", dataTime).put("data", returnData);
        }
    }

    public void timeOutClose(NetSocket netSocket, Map<String, Long> keepLiveMap) {
        String nowTime = TimeUtil.parseTime(LocalDateTime.now(), TimeUtil.TimeFormat.LONG_DATE_PATTERN_LINE);
        String startTime = TimeUtil.parseTime(TimeUtil.getDateTimeOfTimestamp(keepLiveMap.get("alive")));
        logger.info("port:{}==>HandlerID:{}=alive: contection timeout=nowTime:{}==>startTime:{}", port, netSocket.writeHandlerID(), nowTime, startTime);

        netSocket.close();
    }

    public void cancelTimer(List<Long> map) {
        map.forEach(e -> {
            vertx.cancelTimer(e);
        });
    }

    public void closeHandler(NetSocket netSocket, Map<String, List<Long>> timerMap, Map<String, List<Long>> timerOther, SharedData sharedData) {
        String handlerID = netSocket.writeHandlerID();
        netSocket.closeHandler(h -> {
            downLine(handlerID);
            cancelTimer(timerMap.get(handlerID));
            cancelTimer(timerOther.get(handlerID));
            calcCount(sharedData, false);
        });

        netSocket.exceptionHandler(e -> {
            logger.info("-----------------------异常关闭下线 " + "==>port:" + port + "==>" + e.getMessage());
            netSocket.close();
        });
    }


    public void calcCount(SharedData sharedData, boolean isAdd) {
        sharedData.getCounter("mycounter", res -> {
            if (res.succeeded()) {
                Counter counter = res.result();
                if (isAdd) {
                    counter.incrementAndGet(count -> {
                        if (count.succeeded()) {
                            logger.info("add after current total run count: " + count.result() + "---");
                        }
                    });
                } else {
                    counter.decrementAndGet(count -> {
                        if (count.succeeded()) {
                            logger.info("decrement after current total run count: " + count.result() + "---");
                        }
                    });
                }

            }
        });
    }

    protected CallBackCmd getCallBackCmd(SendCmd sendCmd) {
        CallBackCmd callBackCmd = new CallBackCmd();
        callBackCmd.setTime(System.currentTimeMillis());
        callBackCmd.setPort(port);
        callBackCmd.setDeviceHex(sendCmd.getDeviceHex());
        callBackCmd.setSource(sendCmd.getSource());
        callBackCmd.setMessageId(sendCmd.getMessageId());
        callBackCmd.setCmd(sendCmd.getCmd());
        callBackCmd.setSave(sendCmd.isSave());
        callBackCmd.setCmdPK(sendCmd.getCmdPK());
        callBackCmd.setCallbackPlatForm(sendCmd.isCallbackPlatForm());
        callBackCmd.setOnline(sendCmd.isOnline());
        callBackCmd.setDeviceAddr(sendCmd.getDeviceAddr());
        return callBackCmd;
    }

    public void downLine(String handlerID) {
        int count = conectCount.decrementAndGet();
        logger.info("connect close downline port:{}==>current server run total:{}==>handlerID:{}", port, count, handlerID);
        if (count <= 1) {
            lineStatus = false;
            offLineEvent();
            String key = "mod_" + port;
            redis.exists(key, (h) -> {
                if (h.result() > 0) {
                    redis.hset("mod_" + port, "offline", TimeUtil.getCurrentDatetime(), h1 -> {
                    });
                }
            });

        }

    }

    public void upLine(Map<String, Long> keepLiveMap, String handlerID) {
        int count = conectCount.incrementAndGet();
        logger.info("connect open upline port:{}==>current server run total:{}==>handlerID:{}", port, count, handlerID);
        if (count == 1) {
            upLineEvent();
            lineStatus = true;
        }
        keepLiveMap.put("alive", System.currentTimeMillis());
    }

    protected void offLineEvent(String deviceId) {
        eventBus.send(Constant.MOD_OFFLINE_STATUS, returnResult(deviceId, new JsonObject().put("status", 0)));
    }
    protected void offLineEvent() {
        eventBus.send(Constant.MOD_OFFLINE_STATUS, returnResult( new JsonObject().put("status", 0)));
    }

    protected void upLineEvent(String companyId,String deviceId) {
        eventBus.send(Constant.MOD_ONLINE_STATUS, returnResult(companyId,deviceId, new JsonObject().put("status",1)));
    }

    protected void upLineEvent() {
        eventBus.send(Constant.MOD_ONLINE_STATUS, returnResult( new JsonObject().put("status",1)));
    }

    protected void sendFailMessage(SendCmd sendCmd) {
        CallBackCmd callBackCmd = getCallBackCmd(sendCmd);
        sendFailMessage(callBackCmd);
    }

    protected void sendFailMessage(CallBackCmd callBackCmd) {
        if (callBackCmd.isCallbackPlatForm()) {
            eventBus.send(Constant.MOD_CALLBACK, returnResult(JsonObject.mapFrom(callBackCmd)));
        }
    }

    protected CallBackCmd pollCallBackCmd(String cmd_Id, ConcurrentLinkedQueue<CallBackCmd> returnQueue) {
        CallBackCmd ele = null;
        while (returnQueue.size() > 0) {
            ele = returnQueue.poll();
            String cmdPK = ele.getCmdPK();
            logger.info("port:{}==>cmdPK: {}==>resultPK:{}==>returnQueue size:{}", port, cmd_Id, cmdPK, returnQueue.size());
            if (StringUtils.equals(cmdPK.toUpperCase(), cmd_Id.toUpperCase())) {
                break;
            } else {
                sendFailMessage(ele);
                ele = null;
            }

        }
        return ele;
    }

    protected CallBackCmd pollCallBackCmd(String cmd_Id, Map<String, CallBackCmd> returnMap,String deviceId) {
        String key = StringUtils.join(deviceId, "-", cmd_Id);
        CallBackCmd ele = null;
        if (returnMap.containsKey(key)){
            ele = returnMap.get(StringUtils.join(deviceId, "-", cmd_Id));
            returnMap.remove(key);
        }else {
            logger.info("port:{}==>cmdPK: {}==>not find ", port, cmd_Id);
        }
        return ele;
    }

    protected CallBackCmd pollCallBackCmd(Map<String,LinkedList<CallBackCmd>>  callBackMessage,String deviceId,String seq) {
        if (callBackMessage.containsKey(deviceId)) {
            LinkedList<CallBackCmd> callBackCmds = callBackMessage.get(deviceId);
            CallBackCmd callBackCmdResult = null;
            if (callBackCmds.size() > 0) {
                for (int i = 0; i < callBackCmds.size(); i++) {
                    CallBackCmd callBackCmd = callBackCmds.get(i);
                    if (StringUtils.equals(callBackCmd.getCmdPK(), seq)) {
                        callBackCmdResult = callBackCmd;
                        callBackCmds.remove(i);
                    }
                }
                return callBackCmdResult;

            } else {
                logger.info("port:{}==>cmdPK: {}==>not find ", port);
            }
        } else {
            logger.info("port:{}==>cmdPK: {}==>not find ", port);
        }
        return null;
    }


    protected void startCmdConsumer() {
        eventBus.consumer(Constant.ONLY_CMD + port, obj -> {
            JsonObject jsonObject = (JsonObject) obj.body();
            SendCmd sendCmd = JSON.parseObject(jsonObject.toString(), SendCmd.class);
            if (!lineStatus) {
                callFailMessage(sendCmd);
            } else {
                eventBus.publish(Constant.ONLY_CMD_LOCAL + port + sendCmd.getDeviceHex(), jsonObject);
            }

        });
    }


    protected void startSetModel() {
        eventBus.consumer(Constant.MOD_SEND_SET_MODEL + port, obj -> {
            JsonObject json = (JsonObject) obj.body();
            new JsonArray(json.getString("deviceList")).forEach(ele -> {
                devicesInfo.remove(ele);
            });
        });
    }

    protected void startCmdConsumerMap(PriorityBlockingQueue<SendCmd> sendQueue,Map<String,Queue<SendCmd>>  delayedCmd) {
        eventBus.consumer(Constant.ONLY_CMD + port, obj -> {
            SendCmd sendCmd = new Gson().fromJson(obj.body().toString(), SendCmd.class);
            DeviceInfoUdp deviceInfoUdp = devicesInfo.get(sendCmd.getDeviceHex());
            if (deviceInfoUdp==null||!deviceInfoUdp.isLineStatus()) {
                sendCmd.setOnline(false);
                callFailMessage(sendCmd);
            } else if (sendCmd.getInstantMessage()){
                sendQueue.offer(sendCmd);
//                eventBus.publish(Constant.ONLY_CMD_LOCAL + port + sendCmd.getDeviceHex(), jsonObject);
            }else {
                Queue<SendCmd> sendCmds = delayedCmd.get(sendCmd.getDeviceHex());
                LinkedList<SendCmd> sendCmdQueue;
                if (sendCmds==null){
                    sendCmdQueue = new LinkedList<>();
                    sendCmdQueue.add(sendCmd);
                    delayedCmd.put(sendCmd.getDeviceHex(), sendCmdQueue);
                }else {
                    sendCmds.add(sendCmd);
                    delayedCmd.put(sendCmd.getDeviceHex(), sendCmds);
                }
            }

        });
    }

    protected boolean isOffLine(SendCmd sendCmd) {
        Integer integer = lineStatusMap.get(sendCmd.getDeviceHex());
        return isNull(integer);
    }

    protected boolean isNull(Integer integer) {
        return integer==null||integer==0;
    }


    protected void callFailMessage( SendCmd sendCmd ) {
        if (sendCmd.isCallback()) {
            sendFailMessage(sendCmd);
        }
    }

    protected void startConsumer(NetSocket netSocket, List<Long> timerOther, PriorityBlockingQueue<SendCmd> queue, ConcurrentLinkedQueue<CallBackCmd> returnQueue, boolean lineStatus) {
        long periodicId = vertx.setPeriodic(1000, id -> {
            int queueSize = queue.size();
            if (queueSize > 0) {
                SendCmd sendCmd = queue.poll();
                if (netSocket.writeQueueFull()) {
                    if (sendCmd.isCallback()) {
                        logger.info("port:{}==>cmd:{}==>queue count:{}==>queuefull==>HandlerId", port, sendCmd.getCmd(), queueSize, netSocket.writeHandlerID());
                        sendFailMessage(sendCmd);
                    }
                } else {
                    sendCmdCallBack(returnQueue, sendCmd);

                    int currqueueSize = queue.size();
                    String cmd = sendCmd.getCmd();
                    if (sendCmd.getSendType() == 0) {
                        Buffer buffer = Buffer.buffer().appendBytes(Utils.hexStringToByteArray(cmd));
                        netSocket.write(buffer, result -> {
                            if (result.succeeded()) {
                                logger.info("cmd send success byte ==>port:{}==>cmd:{}==>queue count:{}", port, cmd, currqueueSize);
                            } else {
                                logger.info("port:{}==>cmd:{}==>queue count:{}==>cmd send fail-info:{}", port, cmd, currqueueSize, result.cause().getMessage());
                            }
                        });
                    } else if (sendCmd.getSendType() == 1) {
                        netSocket.write(cmd, result -> {
                            if (result.succeeded()) {
                                logger.info("cmd send success String==>port:{}==>cmd:{}==>queue count:{}", port, cmd, currqueueSize);
                            } else {
                                logger.info("port:{}==>cmd:{}==>queue count:{}==>cmd send fail-info:{}", port, cmd, currqueueSize, result.cause().getMessage());
                            }
                        });
                    }
                }


            }


        });

        checkCmdTimeOut(timerOther, returnQueue);
        timerOther.add(periodicId);

        onlyCmdLocal(queue);
    }


    private void checkCmdTimeOut(List<Long> timerOther, Map<String,LinkedList<CallBackCmd>>  callBackMessage) {
        long returnPeriodicId = vertx.setPeriodic(1000*3, id -> {
            callBackMessage.forEach((k, callBackCmdQueue) -> {
                long now = System.currentTimeMillis();
                if(callBackCmdQueue.size()>0){
                    CallBackCmd v = callBackCmdQueue.peek();
                    long consumor = callBackCmdQueue.peek().getTime();
                    if ((now - consumor) > 15 * 1000) {
                        String nowTime = TimeUtil.parseTime(TimeUtil.getDateTimeOfTimestamp(now));
                        String consumorTime = TimeUtil.parseTime(TimeUtil.getDateTimeOfTimestamp(consumor));
                        logger.info("port:{}==>nowTime:{}==>consumorTime:{}==>callBackCmd:{}", port, nowTime, consumorTime, v.toString());
                        cmdCallBackFailCount.put(v.getDeviceHex(), cmdCallBackFailCount.get(v.getDeviceHex())==null?0:cmdCallBackFailCount.get(v.getDeviceHex())+1);
                        cmdCallBackFailCount.forEach((deviceId,failCount) ->{
                                logger.debug("devicdId:{},send  cmd fail count:{} ",deviceId,failCount);
                        });
                        if (v.isCallbackPlatForm()) {
                            sendFailMessage(v);
                        }
                        callBackCmdQueue.poll();
                    }
                }


            });
        });
        timerOther.add(returnPeriodicId);
    }

    private void checkCmdTimeOut(List<Long> timerOther, ConcurrentLinkedQueue<CallBackCmd> returnQueue) {
        long returnPeriodicId = vertx.setPeriodic(1000 * 2, id -> {
            if (returnQueue.size() > 0) {
                CallBackCmd callBackCmd = returnQueue.peek();
                long now = System.currentTimeMillis();
                long consumor = callBackCmd.getTime();
                if ((now - consumor) > 15*1000) {
                    String nowTime = TimeUtil.parseTime(TimeUtil.getDateTimeOfTimestamp(now));
                    String consumorTime = TimeUtil.parseTime(TimeUtil.getDateTimeOfTimestamp(consumor));
                    logger.info("port:{}==>queue count:{}==>nowTime:{}==>consumorTime:{}==>callBackCmd:{}", port, returnQueue.size(), nowTime, consumorTime, callBackCmd.toString());
                    returnQueue.poll();
                    if (callBackCmd.isCallbackPlatForm()) {
                        sendFailMessage(callBackCmd);
                    }
                }
            }

        });
        timerOther.add(returnPeriodicId);
    }

    protected void onlyCmdLocal(PriorityBlockingQueue<SendCmd> queue,String device) {
        eventBus.localConsumer(Constant.ONLY_CMD_LOCAL + port+device, obj -> {
            SendCmd sendCmd = JSON.parseObject(obj.body().toString(), SendCmd.class);
            queue.offer(sendCmd);
        });
    }

    protected void onlyCmdLocal(PriorityBlockingQueue<SendCmd> queue) {
        eventBus.localConsumer(Constant.ONLY_CMD_LOCAL + port, obj -> {
            SendCmd sendCmd = JSON.parseObject(obj.body().toString(), SendCmd.class);
            queue.offer(sendCmd);
        });
    }
    protected void udpStartConsumer(UdpSendCmd udpSendCmd) {
        PriorityBlockingQueue<SendCmd> queue = udpSendCmd.getQueue();
        Map<String,LinkedList<CallBackCmd>>  callBackMessage = udpSendCmd.callBackMessage;
        DatagramSocket socket = udpSendCmd.getSocket();

        long periodicId = vertx.setPeriodic(10, id -> {
            int queueSize = queue.size();
            if (queueSize > 0) {
                SendCmd sendCmd = queue.poll();
                sendCmdCallBack(callBackMessage , sendCmd);
                uddSendCmd(socket, sendCmd, queue.size());
            }
        });
        timerOtherUdp.add(periodicId);

        checkCmdTimeOut(timerOtherUdp, callBackMessage);
    }


    protected void uddSendCmd(DatagramSocket socket, SendCmd sendCmd, int currqueueSize) {
        DeviceInfoUdp udpInfo = devicesInfo.get(sendCmd.getDeviceHex());
        if (udpInfo == null) {
            sendFailMessage(sendCmd);
            return;
        }
        long interval = 100;
        if (currqueueSize >= 0) {
            interval = 500;
        }
        if (System.currentTimeMillis() - udpInfo.getLastSendCmdTime() < interval) {
            vertx.setTimer(interval, h -> {
                uddSendCmd(socket, sendCmd, currqueueSize);
            });
        } else {
            udpInfo.setLastSendCmdTime(System.currentTimeMillis());
            String remoteAddress = udpInfo.getRemoteAddress();
            Integer remotePort = udpInfo.getRemotePort();

            String cmd = sendCmd.getCmd();

            if (sendCmd.getSendType() == 0) {
                Buffer buffer = Buffer.buffer().appendBytes(Utils.hexStringToByteArray(cmd));
                socket.send(buffer, remotePort, remoteAddress, result -> {
                    sendCmdLog(currqueueSize, cmd, result, udpInfo);
                });
            } else if (sendCmd.getSendType() == 1) {
                socket.send(cmd, remotePort, remoteAddress, result -> {
                    sendCmdLog(currqueueSize, cmd, result, udpInfo);
                });
            }
        }

    }

    protected void checkLine(long time, long timeout, List<Long> timerOther) {
        long overtimeId = vertx.setPeriodic(time, t -> {
            devicesInfo.forEach((k, deviceInfoUdp) -> {
                long lastDataTime = deviceInfoUdp.getLastDataTime();
                if (System.currentTimeMillis() - lastDataTime > timeout) {
                    offLineEvent(k);
                    devicesInfo.remove(k);
                    redis.hmset(getRedisKeyByDeviceId(k), new JsonObject().put("lineStatus", false).put("lastDataTime", lastDataTime), h->{

                    });
                }
            });
        });
        timerOther.add(overtimeId);
    }

    private void sendCmdCallBack(ConcurrentLinkedQueue<CallBackCmd> returnQueue, SendCmd sendCmd) {
        if (sendCmd.isCallback()) {
            CallBackCmd callBackCmd = getCallBackCmd(sendCmd);
            returnQueue.offer(callBackCmd);
        }
    }

    private void sendCmdCallBack(Map<String,LinkedList<CallBackCmd>>  callBackMessage, SendCmd sendCmd) {
        if (sendCmd.isCallback()) {
            CallBackCmd callBackCmd = getCallBackCmd(sendCmd);
            LinkedList<CallBackCmd> callBackCmds = callBackMessage.get(callBackCmd.getDeviceHex());
            if (callBackCmds==null){
                callBackCmds=new LinkedList<>();
            }
            callBackCmds.add(callBackCmd);
            callBackMessage.put(callBackCmd.getDeviceHex(), callBackCmds);
        }
    }

    protected void rpcCallSucc(CallBackCmd ele) {
        eventBus.send(Constant.MOD_CALLBACK, returnResult(JsonObject.mapFrom(ele)));
    }

    private void sendCmdLog(int currqueueSize, String cmd, AsyncResult<DatagramSocket> result, DeviceInfoUdp udpInfo) {
        String remoteAddress = udpInfo.getRemoteAddress();
        Integer remotePort = udpInfo.getRemotePort();
        if (result.succeeded()) {
            logger.info("cmd send success ==>port:{}==>remoteAddress:{},remotePort:{}==>cmd:{}==>queue count:{}", port, remoteAddress, remotePort, cmd, currqueueSize);
        } else {
            logger.info("port:{}==>remoteAddress:{},remotePort:{}==>cmd:{}==>queue count:{}==>cmd send fail-info:{}", port, remoteAddress, remotePort, cmd, currqueueSize, result.cause().getMessage());
        }
    }

    protected void rpcCallBack(byte[] bytes, CallBackCmd ele,Object data,boolean succ,Integer status) {
        if (ele != null) {
            ele.setStatus(status);
            ele.setSuccess(succ);
            if (data!=null){
                ele.setData(data);
            }
            ele.setCallBackCmd(Utils.bytes2String(bytes));
            rpcCallSucc(ele);
        }
    }

    private void sendCmdLog(String cmd, AsyncResult<DatagramSocket> result, DeviceInfoUdp udpInfo) {
        String remoteAddress = udpInfo.getRemoteAddress();
        Integer remotePort = udpInfo.getRemotePort();
        if (result.succeeded()) {
            logger.info("cmd send success ==>port:{}==>remoteAddress:{},remotePort:{}==>cmd:{}==>queue count:{}", port, remoteAddress, remotePort, cmd);
        } else {
            logger.info("port:{}==>remoteAddress:{},remotePort:{}==>cmd:{}==>queue count:{}==>cmd send fail-info:{}", port, remoteAddress, remotePort, cmd, result.cause().getMessage());
        }
    }

    protected <T> T getValueBytype(String data, int i, int j, Class<T> clazz) {
        if (int.class == clazz || Integer.class == clazz) {
            return (T) Integer.valueOf(StrUtils.getSmall(getSubstring(data, i, j)), 16);
        } else if (String.class == clazz) {
            return (T) StrUtils.getSmall(getSubstring(data, i, j));
        } else {
            return (T) StrUtils.getSmall(getSubstring(data, i, j));
        }
    }

    protected <T> T getValueBytype(byte[] data, int i, int j, Class<T> clazz) {
        if (int.class == clazz || Integer.class == clazz) {
            return (T) Utils.bytestointerger(data, i, j);
        } else if (String.class == clazz) {
            return (T) Utils.bytes2String(data, i, j);
        }else if (Long.class == clazz) {
            return (T) Utils.toLong(Utils.getbytes(data, i, j), 0);
        } else {
            return (T) Utils.bytes2String(data, i, j);
        }
    }


    protected <T> T getValueBytype(byte[] data, int i, Class<T> clazz) {
        return getValueBytype(data, i, 1, clazz);
    }

    protected Integer getValueBytype(byte[] data, int i) {
        return getValueBytype(data, i, 1, Integer.class);
    }

    protected Integer getValueBytype(byte[] data, int i, int j) {
        return getValueBytype(data, i, j, Integer.class);
    }
    protected void saveMongoJson(List<BulkOperation> bulkOperationList, JsonArray lasteDat, double value, int type) {
        saveMongoJson(bulkOperationList, lasteDat, null, value, type);
    }


    protected void saveMongoJson(List<BulkOperation> bulkOperationList, JsonObject baseJson, int value, int type) {
        Map<Integer, ParamsSetting> paramsSettingAll = paramsSettingService.findAll();
        ParamsSetting paramsSetting = paramsSettingAll.get(type);
        baseJson.put("value", value);
        baseJson.put("type", type);
        if (paramsSetting == null) {
            baseJson.put("unit", "");
            baseJson.put("des", "");
        } else {
            baseJson.put("unit", paramsSetting.getUnit());
            baseJson.put("des", paramsSetting.getDes());
        }
        String currentTime = TimeUtil.getCurrentDatetime(TimeUtil.TimeFormat.LONG_DATE_PATTERN_WITH_MILSEC_NONE_Z);
        baseJson.put("port", port);
        baseJson.put("createdAt", new JsonObject().put("$date", currentTime));
        bulkOperationList.add(BulkOperation.createInsert(baseJson));
    }

    protected void saveMongoJson(List<BulkOperation> bulkOperationList, JsonArray lasteDate, JsonObject baseJson, int value, int type) {
        Map<Integer, ParamsSetting> paramsSettingAll= paramsSettingService.findAll();
        ParamsSetting paramsSetting = paramsSettingAll.get(type);
        JsonObject entity;
        if (baseJson != null) {
            entity = baseJson.copy();
        } else {
            entity = new JsonObject();
        }
        entity.put("value", value);
        entity.put("type", type);
//        entity.put("deviceId", "0001");
        if (paramsSetting == null) {
//            logger.error(type + "=================================");
            entity.put("unit", "");
            entity.put("des", "");
        } else {
            entity.put("unit", paramsSetting.getUnit());
            entity.put("des", paramsSetting.getDes());
        }
        String currentTime = TimeUtil.getCurrentDatetime(TimeUtil.TimeFormat.LONG_DATE_PATTERN_WITH_MILSEC_NONE_Z);
        JsonObject saveJson = entity.copy();
        lasteDate.add(entity);
        saveJson.put("port", port);
        saveJson.put("createdAt", new JsonObject().put("$date", currentTime));
        bulkOperationList.add(BulkOperation.createInsert(saveJson));
    }

    protected void saveMongoJson(List<BulkOperation> bulkOperationList, JsonArray lasteDate, JsonObject baseJson, double value, int type) {
        Map<Integer, ParamsSetting> paramsSettingAll= paramsSettingService.findAll();
        ParamsSetting paramsSetting = paramsSettingAll.get(type);
        JsonObject entity;
        if (baseJson != null) {
            entity = baseJson.copy();
        } else {
            entity = new JsonObject();
        }
        entity.put("value", value);
        entity.put("type", type);
//        entity.put("deviceId", "0001");
        if (paramsSetting == null) {
//            logger.error(type + "=================================");
            entity.put("unit", "");
            entity.put("des", "");
        } else {
            entity.put("unit", paramsSetting.getUnit());
            entity.put("des", paramsSetting.getDes());
        }
        String currentTime = TimeUtil.getCurrentDatetime(TimeUtil.TimeFormat.LONG_DATE_PATTERN_WITH_MILSEC_NONE_Z);
        JsonObject saveJson = entity.copy();
        lasteDate.add(entity);
        saveJson.put("port", port);
        saveJson.put("createdAt", new JsonObject().put("$date", currentTime));
        bulkOperationList.add(BulkOperation.createInsert(saveJson));
    }

    protected void getPolice(JsonObject entity, JsonArray policeDate,int type,Object value) {
        Map<Integer, ExceColleContr> exceColleal = exceColleContrService.findAll();
        ExceColleContr exceColleContr = exceColleal.get(type);
        JsonObject police;
        if (entity!=null){
            police = entity.copy();
        }else {
            police=new JsonObject();
        }
        police.put("port", port);
        police.put("type", type);
        if (value!=null){
            police.put("value", value);
        }
        if(exceColleContr!=null){
            police.put("unit", exceColleContr.getExceptions_des());
        }
        policeDate.add(police);
    }

    protected void getPolice(JsonObject entity,JsonArray policeDate,int type) {
        getPolice(entity,policeDate,type,null);
    }

    protected void getPolice(JsonArray policeDate,int type,Object value) {
        getPolice(null,policeDate,type,value);
    }

    protected void getPolice(JsonArray policeDate,int type) {
        getPolice(null,policeDate,type,null);
    }

    protected   boolean isPolice(Long CI_Warn,long policeValue) {
        return (CI_Warn & policeValue) == policeValue;
    }

    protected   boolean isPolice(int CI_Warn,int policeValue) {
        return (CI_Warn & policeValue) == policeValue;
    }

    public static String getRedisKeyByDeviceId(String deviceId) {
        return "device-"+deviceId;
    }

    protected void saveMongoAndSend(String dataName, MongoClient mongoClient, List<BulkOperation> bulkOperationList, EventBus eventBus, JsonArray lasteDate) {
        mongoClient.bulkWrite(dataName, bulkOperationList, stringAsyncResult -> {
            if (stringAsyncResult.succeeded()) {
                eventBus.send(Constant.MOD_LATEST, returnResult(lasteDate.toString()));
            } else {
                PublicUtils.printException(stringAsyncResult.cause(), port);
            }
        });
    }
    protected void saveMongoAndSend(List<BulkOperation> bulkOperationList, JsonArray lasteDate) {
        mongoClient.bulkWrite(Constant.MONGONAME_REAL, bulkOperationList, stringAsyncResult -> {
            if (stringAsyncResult.failed()) {
                PublicUtils.printException(stringAsyncResult.cause(), port);
            } else {
            }
        });
    }

    protected void saveMongoAndSend(String dataName, MongoClient mongoClient, List<BulkOperation> bulkOperationList, EventBus eventBus, JsonArray lasteDate, String deviceId) {
        mongoClient.bulkWrite(dataName, bulkOperationList, stringAsyncResult -> {
            if (stringAsyncResult.succeeded()) {
//                logger.info("save count:" +saveintc.incrementAndGet());
                eventBus.send(Constant.MOD_LATEST, returnResult(deviceId,lasteDate.toString()));
//                sendDataToMq(returnResult(deviceId,lasteDate));

            } else {
                PublicUtils.printException(stringAsyncResult.cause(), port);
            }
        });
    }

    protected void saveMongo(String dataName,List<BulkOperation> bulkOperationList) {
        mongoClient.bulkWrite(dataName, bulkOperationList, stringAsyncResult -> {
            if (stringAsyncResult.succeeded()) {
            } else {
                PublicUtils.printException(stringAsyncResult.cause(), port);
            }
        });
    }

    protected void sendDataToMq(JsonObject data){
        Message msg = new Message( //
                // Message所属的Topic
                mqConfig.getOrderTopic(),
                // Message Tag 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在MQ服务器过滤
                mqConfig.getOrderTag(),
                // Message Body 可以是任何二进制形式的数据， MQ不做任何干预
                // 需要Producer与Consumer协商好一致的序列化和反序列化方式
                data.toString().getBytes());
        // 设置代表消息的业务关键属性，请尽可能全局唯一
        // 以方便您在无法正常收到消息情况下，可通过MQ 控制台查询消息并补发
        // 注意：不设置也不会影响消息正常收发
        msg.setKey(StringUtils.join(mqConfig.getOrderTopic(),"-",mqConfig.getOrderTag(),"-",data.hashCode()));
        // 发送消息，只要不抛异常就是成功
        try {
            producer.sendAsync(msg, new SendCallback() {
                @Override
                public void onSuccess(final SendResult sendResult) {
                    assert sendResult != null;
                    logger.info(sendResult.toString());
                }

                @Override
                public void onException(final OnExceptionContext context) {
                    logger.error("send mq fail message:{}",context);
                    //出现异常意味着发送失败，为了避免消息丢失，建议缓存该消息然后进行重试。
                }
            });
        } catch (ONSClientException e) {
            System.out.println("发送失败");
            //出现异常意味着发送失败，为了避免消息丢失，建议缓存该消息然后进行重试。
        }
    }

    protected String getSubstring(String data, int i, int j) {
        return data.substring(i, i + j);
    }

    protected String parseDataStr(String datas, int lengthPosition, int lenLen, Consumer<String> consumer) {
        int totalLength;
        String subData;
        while (true) {
            int dataLenth = datas.length();
            if (dataLenth > lengthPosition + lenLen) {
                totalLength = getValueBytype(datas, lengthPosition, lenLen, int.class) * 2;
                if (totalLength > 1000) {
                    subData = "";
                    break;
                }
            } else {
                subData = datas;
                break;
            }
            if (dataLenth < totalLength) {
                subData = datas;
                logger.info("port:{}==>dataLenth:{}==>totalLength:{} string data lenth not enough:{}", port, dataLenth, totalLength, datas);
                break;
            } else if (dataLenth == totalLength) {
                consumer.accept(datas);
                subData = "";
                break;
            } else {
                String preSub = datas.substring(0, totalLength);
                datas = datas.substring(totalLength);
                consumer.accept(preSub);
            }

        }

        return subData;

    }

    protected Buffer parseDataByte(byte[] datas, int lengthPosition, int lenLen, int otherLen, Consumer<byte[]> consumer) {
//        byte[] lengFlag; // 标识符byte数组
//        int lengthPosition; // 长度
//        int other; // 其他长度
//        boolean gotoFlag=false;
//        int spitPosition = 0; // 第一次截取位置
//        byte[] lengFlag; // 标识符byte数组
//        int lengthPosition; // 长度
//        int other; // 其他长度
//        boolean gotoFlag=false;

        int totalLength;
        byte[] subData;
        while (true) {
            int dataLenth = datas.length;
            if (dataLenth > lengthPosition + lenLen) {
                totalLength = Utils.bytestoint(datas, lengthPosition, lenLen) + otherLen;
                if (totalLength > 1000) {
                    subData = new byte[0];
                    break;
                }
            } else {
                subData = datas;
                break;
            }
            if (dataLenth < totalLength) {
                subData = datas;
                logger.info("port:{} byte data lenth not enough", port);
                break;
            } else if (dataLenth == totalLength) {
                consumer.accept(datas);
                subData = new byte[0];
                break;
            } else {
                byte[] preSub = Arrays.copyOfRange(datas, 0, totalLength);
                datas = Arrays.copyOfRange(datas, totalLength, dataLenth);
                consumer.accept(preSub);
            }

        }

        return Buffer.buffer(subData);
    }

    protected Buffer parseCan(byte[] bytes, Buffer preData, Map<String, byte[]> preBytes, byte[] preDataBytes, byte[] tailDataBytes, Consumer<byte[]> consumer) {
        int totalLength = 13;
        while (true) {
            int dataLenth = bytes.length;
//            logger.debug(dataLenth + "------" + Utils.bytes2String(preData.getBytes()));
            if (dataLenth < totalLength) {
                preData = Buffer.buffer(bytes);
                break;
            } else if (dataLenth == totalLength) {
                parseCanContent(bytes, preBytes, preDataBytes, tailDataBytes, consumer);
                preData = Buffer.buffer();
                break;
            } else if (dataLenth > totalLength) {
                preData = Buffer.buffer();
                byte[] datas = Arrays.copyOfRange(bytes, 0, 13);
                bytes = Arrays.copyOfRange(bytes, 13, dataLenth);
                parseCanContent(datas, preBytes, preDataBytes, tailDataBytes, consumer);
            }

        }
        return preData;
    }

    protected void parseCanContent(byte[] canData, Map<String, byte[]> prebytesMap, byte[] preDataBytes, byte[] tailDataBytes, Consumer<byte[]> consumer) {
        int len = Utils.bytestoint(canData, 0);
        if (len > 8) {
            prebytesMap.put("prebytes", new byte[0]);
            logger.error("port:{},invalid data:{}", port, Utils.bytes2String(canData));
            return;
        }
        byte[] preBytes = prebytesMap.get("prebytes");
        canData = Arrays.copyOfRange(canData, 5, 5 + len);
        if (preBytes != null && preBytes.length > 0) {
//            logger.debug(Utils.bytes2String(preBytes));
            canData = Bytes.concat(preBytes, canData);
        }
        int preExists = Bytes.indexOf(canData, preDataBytes);
        int tailExists = Bytes.indexOf(canData, tailDataBytes);
        if (preExists >= 0) {
             if (tailExists >= 0) {
                preBytes = Arrays.copyOfRange(canData, tailExists + 2, canData.length);
                if (tailExists == canData.length - 2){
                    prebytesMap.put("prebytes", new byte[0]);
                }else {
                    prebytesMap.put("prebytes", preBytes);
                }
                consumer.accept(Arrays.copyOfRange(canData, preExists, tailExists + 2));
            } else {
                prebytesMap.put("prebytes", canData);
            }
        } else {
            prebytesMap.put("prebytes", new byte[0]);
            logger.error("port:{},invalid data:{}", port, Utils.bytes2String(canData));
        }
    }
    protected class UdpSendCmd{
        @Setter
        @Getter
        private DatagramSocket socket;
        @Setter
        @Getter
        private PriorityBlockingQueue<SendCmd> queue;
        @Setter
        @Getter
        private Map<String,LinkedList<CallBackCmd>> callBackMessage;
        @Setter
        @Getter
        private Map<String,Queue<SendCmd>>  delayedCmd;

        public UdpSendCmd(DatagramSocket socket, PriorityBlockingQueue<SendCmd> queue, Map<String,LinkedList<CallBackCmd>> callBackMessage, Map<String, Queue<SendCmd>> delayedCmd) {
            this.socket = socket;
            this.queue = queue;
            this.callBackMessage = callBackMessage;
            this.delayedCmd = delayedCmd;
        }
    }

    @Data
    public class DeviceInfoUdp {
        private String remoteAddress;
        private Integer remotePort;
        private boolean lineStatus =true;
        private long lastDataTime ;
        private long lastCurrentDataTime =0;
        private String companyId ;
        private String deviceId  ;
        private Integer currentRatio =1;
        private String source;
        private String hardwareType ;
        private long lastSendCmdTime ;


        public DeviceInfoUdp(String remoteAddress, Integer remotePort, boolean lineStatus) {
            this.remoteAddress = remoteAddress;
            this.remotePort = remotePort;
            this.lineStatus = lineStatus;
        }

        public DeviceInfoUdp(String remoteAddress, Integer remotePort) {
            this.remoteAddress = remoteAddress;
            this.remotePort = remotePort;
        }
    }
}
